package com.allen.flink.batch.connector;

import org.apache.flink.addons.hbase.TableInputFormat;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.ConfigConstants;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * 功能: flink 读取HBase
 *
 * @date: 2020-03-26 11:25
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class ReadFromHBase {
    
    public final static byte[] family = "F".getBytes(ConfigConstants.DEFAULT_CHARSET);
    
    public static void main(String[] args) throws Exception {
        //获取执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        //读取数据
        DataSet<Tuple4<String, String, Integer, String>> users = env.createInput(new TableInputFormat<Tuple4<String, String, Integer, String>>() {
            
            @Override
            public void configure(final Configuration config) {
                
                table = createTable();
                if (null != table) {
                    scan = getScanner();
                }
            }
            
            private HTable createTable() {
                org.apache.hadoop.conf.Configuration config = HBaseConfiguration.create();
                config.set("hbase.zookeeper.quorum", "master,slave1,slave2");
                config.set("hbase.zookeeper.property.clientPort", "2181");
                try {
                    return new HTable(config, getTableName());
                } catch (IOException e) {
                    e.printStackTrace();
                }
                return null;
            }
            
            @Override
            protected Scan getScanner() {
                Scan scan = new Scan();
                scan.addFamily(family);
                return scan;
            }
            
            @Override
            protected String getTableName() {
                return "learing:users";
            }
            
            @Override
            protected Tuple4<String, String, Integer, String> mapResultToTuple(final Result result) {
                
                return Tuple4.of(Bytes.toString(result.getRow()), Bytes.toString(result.getValue(family, "name".getBytes(ConfigConstants.DEFAULT_CHARSET))),
                    Integer.parseInt(Bytes.toString(result.getValue(family, "age".getBytes(ConfigConstants.DEFAULT_CHARSET)))),
                    Bytes.toString(result.getValue(family, "address".getBytes(ConfigConstants.DEFAULT_CHARSET))));
            }
        });
        
        users.print();
    }
}
